package rdd

import org.apache.spark.{SparkConf, SparkContext}

object RDD_Operation {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName("RDD_Operation")

    val sc = new SparkContext(conf)

    // 构造两个RDD
    val rdd1 = sc.parallelize(List(1, 2, 3, 3))
    val rdd2 = sc.parallelize(List(3, 4, 5))
    println(rdd1.collect().mkString(","))
    println(rdd2.collect().mkString(","))
    println("*" * 100)

    // union：合并两个集合
    val rdd3 = rdd1.union(rdd2)
    println(rdd3.collect().mkString(","))
    println("*" * 100)

    // intersection：交集。只保留在两个rdd中都有的元素
    val rdd4 = rdd1.intersection(rdd2)
    println(rdd4.collect().mkString(","))
    println("*" * 100)

    // subtract：差集。只保留在第一个rdd中有而在第二个rdd中没有的元素
    val rdd5 = rdd1.subtract(rdd2)
    println(rdd5.collect().mkString(","))
    println(rdd2.subtract(rdd1).collect().mkString(","))
    println("*" * 100)

    // cartesian：两个rdd的笛卡尔集；当在类型为T和U的RDD上调用时，返回一个(T, U)对(所有元素对)的RDD
    val rdd6 = rdd1.cartesian(rdd2)
    println(rdd6.collect().mkString(","))
    println("*" * 100)

    // zip：拉链；当在类型为T和U的RDD上调用时，返回一个(T, U)对的RDD，其中元组第一个元素来自第一个RDD，第二个元素来自第二个RDD
    // 两个RDD具有相同数量的分区和每个分区中相同数量的元素
    val rddA = sc.parallelize(Array("aa", "bb", "cc", "dd"), 3)
    val rddB = sc.parallelize(Array(1, 2, 3, 4), 3)
    val rddC = rddA.zip(rddB)
    println(rddC.collect().mkString(","))


    sc.stop()
  }
}
